package com.cloud.message.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQListener;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * RocketMQ消息消费，解决每个队列消费顺序的问题
 *
 * @author fengwenjin
 */
@Slf4j
public class RocketMQConsumer implements MessageListenerConcurrently {
    private RocketMQListener rocketMQListener;

    public RocketMQConsumer(RocketMQListener rocketMQListener) {
        this.rocketMQListener = rocketMQListener;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt messageExt : msgs) {
            log.debug("received msg: {}", messageExt);
            try {
                long now = System.currentTimeMillis();
                Future<?> future = ThreadPoolProcessor.getExecutor(context.getMessageQueue()).submit(() -> {
                    rocketMQListener.onMessage(new String(messageExt.getBody(), Charset.forName("UTF-8")));
                });
                future.get();
                long costTime = System.currentTimeMillis() - now;
                log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
            } catch (Exception e) {
                log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);
                context.setDelayLevelWhenNextConsume(0);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    /**
     * 线程池内部类
     */
    static class ThreadPoolProcessor {
        static List<ThreadPoolExecutor> executorList = new ArrayList<>();

        static {
            int processorCount = Runtime.getRuntime().availableProcessors() * 2;
            for (int i = 0; i < processorCount; i++) {
                ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
                executorList.add(threadPoolExecutor);
            }
        }

        static ThreadPoolExecutor getExecutor(MessageQueue queue) {
            String key = queue.getTopic() + queue.getQueueId();
            int hashCode = key.hashCode();
            int executorSize = executorList.size();
            int index = hashCode > 0 ? (hashCode % executorSize) : (Math.abs(hashCode) % executorSize);
            return executorList.get(index);
        }
    }
}
